package com.zyh.sqc.service.impl;

import com.zyh.sqc.service.StateQueryService;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.queryablestate.client.QueryableStateClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.net.UnknownHostException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@Service
public class StateQueryServiceImpl implements StateQueryService {
    @Value("${flink.queryable.state.remoteHost}")
    private String remoteHost;
    @Value("${flink.queryable.state.remotePort}")
    private int remotePort;
    @Value("${flink.queryable.state.jobId}")
    private String jobId;
    @Value("${flink.queryable.state.queryableName}")
    private String queryableName1;

    @Override
    public String queryReport(String appName, String username, String uuid) {
        //提供taskManager的ip和port
        QueryableStateClient client = null;
        try {
            client = new QueryableStateClient(remoteHost, remotePort);
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }

        //准备4方面的参数
        JobID jobID = JobID.fromHexString(jobId);//jobID
        String queryableName = queryableName1;//状态查询名
        String key = appName+":"+username;//键
        TypeInformation<String> keyInformation = Types.STRING;//键的类型
        MapStateDescriptor<String, String> msd = new MapStateDescriptor<>("msd", Types.STRING, Types.STRING);//状态描述符

        CompletableFuture<MapState<String, String>> future = client.getKvState(jobID, queryableName, key, keyInformation, msd);
        MapState<String, String> mapState = null;
        try {
            mapState = future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }
        String reportJson = null;
        try {
            reportJson = mapState.get(uuid);
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("reportJson = " + reportJson);
        client.shutdownAndWait();
        return reportJson;
    }
}
